package com.zhang.spark_1.spark_core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title:
 * @author: zhang
 * @date: 2021/12/8 23:05 
 */
object Spark05_req2_top10 {

  def main(args: Array[String]): Unit = {
    //获取spark的连接
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("top10")
    val sc: SparkContext = new SparkContext(conf)
    //读取数据
    val rdd: RDD[String] = sc.textFile("datas/user_visit_action.txt")
    rdd.cache()


    val top10Category: Array[String] = getHotCategory(rdd)

    val clickRDD: RDD[String] = rdd.filter(
      line => {
        val datas: Array[String] = line.split("_")
        if (datas(6) != "-1") {
          top10Category.contains(datas(6))
        } else {
          false
        }
      }
    )

    //根据品类ID和sessionid进行统计（（品类ID，sessionID）），1）
    val reduceMap: RDD[((String, String), Int)] = clickRDD.map(
      line => {
        val datas: Array[String] = line.split("_")
        ((datas(6), datas(2)), 1)
      }
    ).reduceByKey(_ + _)
    val mapRDD: RDD[(String, (String, Int))] = reduceMap.map {
      case ((cid, sid), sum) => (cid, (sid, sum))
    }

    val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD.groupByKey()

    val resultRDD: RDD[(String, List[(String, Int)])] = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    )

    resultRDD.foreach(println)
    sc.stop()
  }


  def getHotCategory(rdd:RDD[String])={
    val flatRDD: RDD[(String, (Int, Int, Int))] = rdd.flatMap(

      line => {
        val datas = line.split("_")
        if (datas(6) != "-1") {
          List((datas(6), (1, 0, 0)))
        } else if (datas(8) != "null") {
          val ids: Array[String] = datas(8).split(",")
          ids.map((_, (0, 1, 0)))
        } else if (datas(10) != "null") {
          val ids: Array[String] = datas(10).split(",")
          ids.map((_, (0, 0, 1)))
        } else {
          Nil
        }
      }
    )
    val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey((t1, t2) => (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3))
    analysisRDD.sortBy(_._2, false).take(10).map(_._1)
  }

}
